package cn.bone.zk.zk03;

import com.alibaba.fastjson.JSON;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

public class DistributedLock {

    private ZooKeeper zkClient;
    private CountDownLatch countDownLatch = new CountDownLatch(1);
    private CountDownLatch waitLatch = new CountDownLatch(1);
    private String waitPath;
    private String currentMode;

    private Logger logger = LoggerFactory.getLogger(DistributedLock.class);

    public DistributedLock(){
        // 获取连接
        String connectString = "192.168.1.5:2001,192.168.1.5:2002,192.168.1.5:2003";
        int sessionTimeout = 2000;
        try {
            zkClient = new ZooKeeper(connectString, sessionTimeout, new Watcher() {
                @Override
                public void process(WatchedEvent event) {

                    // countDownLatch 如果连接上zk  可以释放
                    if(event.getState() == Event.KeeperState.SyncConnected){
                        countDownLatch.countDown();
                    }

                    // waitLatch
                    // 如果事件的类型是删除节点，且删除的节点是上一个节点
                    if(event.getType() == Event.EventType.NodeDeleted
                            && event.getPath().equals(waitPath)){
                        waitLatch.countDown();
                    }
                }
            });
        } catch (IOException e) {
            e.printStackTrace();
        }

        // 等待zk正常连接后，往下走程序
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


        // 判断根节点是否存在
        try {
            Stat stat = zkClient.exists("/locks", false);
            // 根节点不存在，创建根节点（永久节点）
            if(stat==null){
                zkClient.create("/locks",
                        "locks".getBytes(),
                        ZooDefs.Ids.OPEN_ACL_UNSAFE,
                        CreateMode.PERSISTENT);
            }

        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }

    // 使用zk加锁
    public void zkLock(){
        // 创建临时带序号的节点
        try {
            currentMode = zkClient.create("/locks/seq-",
                    null,
                    ZooDefs.Ids.OPEN_ACL_UNSAFE,
                    CreateMode.EPHEMERAL_SEQUENTIAL);

            // 判断创建的节点是否是最小序号的节点，如果是获取到锁，如果不是监听他序号前一个节点
            List<String> children = zkClient.getChildren("/locks", false);
            logger.info(JSON.toJSONString(children));
            // 如果children只有一个值，那就直接获取锁；
            // 如果有多个节点，需要判断，谁最小
            if(children.size()==1){
                return;
            }
            Collections.sort(children);
            // 字符串截取，获得创建子节点的路径 seq-0000000001
            String node = currentMode.substring("/locks/".length());
            logger.info(node);
            // 通过子节点的名字，获取子节点在children集合中的位置
            int index = children.indexOf(node);
            if(index == -1){
                logger.error("数据异常");
                return;
            }
            // 当前节点是第一个节点， 获取到了锁
            if(index==0){
                return;
            }
            // 获取当前子节点的上一个节点
            waitPath = "/locks/" + children.get(index-1);
            zkClient.getData(waitPath, true, null);

            // 等待监听
            waitLatch.await();

        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }


    }

    // 使用zk解锁
    public void zkUnlock(){
        try {
            // 删除节点
            zkClient.delete(currentMode, -1);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }
}
